package com.fwmagic.flink.projectcase.queryactivitycase;

import com.fwmagic.flink.projectcase.queryactivitycase.async.AsyncGPSToActivityBeanFunction;
import com.fwmagic.flink.projectcase.queryactivitycase.bean.ActivityBean;
import com.fwmagic.flink.projectcase.queryactivitycase.utils.FlinkUtilsV1;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;

import java.util.concurrent.TimeUnit;

public class AsyncQueryProvinceByGPSSInkMysqlApplication {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> kafkaSource = FlinkUtilsV1.createKafkaSource(args, new SimpleStringSchema(), env);

        //异步查询出province,得到beans
        SingleOutputStreamOperator<ActivityBean> beans = AsyncDataStream.unorderedWait(kafkaSource, new AsyncGPSToActivityBeanFunction(), 1000, TimeUnit.MILLISECONDS, 10);

        //SingleOutputStreamOperator<ActivityBean> sumed1 = beans.keyBy("aid", "eventType").sum("counts");

        SingleOutputStreamOperator<ActivityBean> sumed2 = beans.keyBy("aid", "eventType", "province").sum("counts");

        //sink to mysql
        //sumed1.addSink(new MySqlSink());

        //sink to redis
        //添加redis配置
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setDatabase(2).build();
        sumed2.addSink(new RedisSink<ActivityBean>(conf, new RedisActivityBeanMapper()));

        env.execute("AsyncQueryProvinceByGPSSInkMysqlApplication");
    }

    public static class RedisActivityBeanMapper implements RedisMapper<ActivityBean> {

        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "ACT_COUNT");
        }

        @Override
        public String getKeyFromData(ActivityBean bean) {
            return bean.aid + "_" + bean.eventType + "_" + bean.province;
        }

        @Override
        public String getValueFromData(ActivityBean bean) {
            return String.valueOf(bean.counts);
        }
    }
}
